package com.dms.producer;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import com.dms.conf.ConfigLoader;
import com.dms.conf.DmsConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DmsProducer<K, V> {

    private static final Logger logger = LoggerFactory.getLogger(DmsProducer.class);
    private Producer<K, V> producer;

    public DmsProducer(String path) throws IOException {
        Properties props = new Properties();
        InputStream in = null;
        try {
            in = new BufferedInputStream(new FileInputStream(path));
            props.load(in);
            producer = new KafkaProducer<K, V>(props);
        } catch (IOException e) {
            logger.error("fail to read config file. {}", e);
        } finally {
            if (in != null) {
                in.close();
            }
        }

    }

    public DmsProducer() throws IOException {
        Properties props;
        props = ConfigLoader.loadFromClasspath(DmsConfig.CONFIG_PRODUCER_FILE_NAME);
        producer = new KafkaProducer<K, V>(props);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     */
    public void produce(String topic, Integer partition, K key, V data) {
        produce(topic, partition, key, data, null, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param timestamp timestamp
     */
    public void produce(String topic, Integer partition, K key, V data, Long timestamp) {
        produce(topic, partition, key, data, timestamp, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param callback  callback
     */
    public void produce(String topic, Integer partition, K key, V data, Callback callback) {
        produce(topic, partition, key, data, null, callback);
    }

    public void produce(String topic, V data) {
        produce(topic, null, null, data, null, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param timestamp timestamp
     * @param callback  callback
     */
    public void produce(String topic, Integer partition, K key, V data, Long timestamp, Callback callback) {
        ProducerRecord<K, V> kafkaRecord = timestamp == null ? new ProducerRecord<K, V>(topic, partition, key, data) : new ProducerRecord<K, V>(topic, partition, timestamp, key, data);
        produce(kafkaRecord, callback);
    }

    public void produce(ProducerRecord<K, V> kafkaRecord) {
        produce(kafkaRecord, (Callback) null);
    }

    public void produce(ProducerRecord<K, V> kafkaRecord, Callback callback) {
        producer.send(kafkaRecord, callback);
    }

    public void close() {
        producer.close();
    }
}
